package day01;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 批处理 word count
 *
 * @author lvbingbing
 * @date 2021-09-01 00:15
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1、创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 2、从文件中读取数据
        DataSet<String> dataSet = env.readTextFile("input/hello.txt");
        // 3、扁平化。对单词进行分组，然后用sum进行聚合
        DataSet<Tuple2<String, Integer>> wordCountDataSet = dataSet.flatMap(new MyFlatMapper())
                .groupBy(0)
                .sum(1);
        // 4、第二种写法
        DataSet<Tuple2<String, Integer>> wordCountDataSet1 = dataSet.flatMap((String lineStr, Collector<Tuple2<String, Integer>> out) -> {
                    String[] words = lineStr.split(" ");
                    for (String word : words) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT))
                .groupBy(0).sum(1);
        // 5、打印输出
        wordCountDataSet.print();
        System.out.println("-----第二种写法-----");
        wordCountDataSet1.print();
    }
}
